package com.shujia;

import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.exec.UDFArgumentLengthException;
import org.apache.hadoop.hive.ql.exec.UDFArgumentTypeException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDF;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDTF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.StructField;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
import org.apache.hadoop.io.Text;
import org.json.JSONArray;
import org.json.JSONException;
import org.json.JSONObject;

import java.util.ArrayList;
import java.util.List;

/**
 * 建表语句：
 * CREATE TABLE moives (movie STRING COMMENT '电影字符串JSON格式');
 * <p>
 * load data local inpath '/usr/local/soft/data/UDTF.txt' into table moives;
 * 数据：
 * {"movie": [{"movie_name": "肖申克的救赎", "MovieType": "犯罪" }, {"movie_name": "肖申克的救赎", "MovieType": "剧情" }]}
 * get_json_object HIVE自带解析Json的函数用法:
 * ①根路径用 $表示
 * ②子路径用 .表示
 * ③取JSON数组中某个JSON 可以用 [下标] 获取
 * ④ 取JSON中Key对应Value,那么可以使用.Key名称获取
 * <p>
 * <p>
 * SELECT get_json_object(movie,'$.movie')  FROM moives
 * SELECT get_json_object(movie,'$.movie[0].movie_name')  FROM moives;
 * <p>
 * 将JSON中的movie 数据取出后为一个JSON数组，要求将数组中的数据转换为多行数据
 * SELECT explodeMovie(get_json_object(movie,'$.movie'))  FROM moives
 * 返回结果为：
 * 肖申克的救赎 犯罪
 * 肖申克的救赎 剧情
 *
 *
 * add jar /usr/local/soft/jars/hiveCode-1.0.jar;
 *
 * create temporary function explodeMovie as 'com.shujia.MovieUDTF';
 *
 */
public class MovieUDTF extends GenericUDTF {

    @Override
    public StructObjectInspector initialize(StructObjectInspector argOIs)
            throws UDFArgumentException {

        List<? extends StructField> allStructFieldRefs = argOIs.getAllStructFieldRefs();
        // 参数长度为 2
        if (allStructFieldRefs.size() != 1) {
            throw new UDFArgumentLengthException("当前参数的长度要求为1个");
        }

        // 抛出类型异常
        if (!allStructFieldRefs.get(0)
                .getFieldObjectInspector()
                .getCategory()
                .equals(ObjectInspector.Category.PRIMITIVE)) {
            throw new UDFArgumentTypeException( 1, "当前第1个参数类型不匹配");
        }

        // 设定返回列的名称以及类型
        ArrayList<String> fieldNames = new ArrayList<String>();
        fieldNames.add("movie_name");
        fieldNames.add("movie_type");

        // 对应列的类型
        ArrayList<ObjectInspector> fieldOIs = new ArrayList<ObjectInspector>();
        fieldOIs.add(PrimitiveObjectInspectorFactory.writableStringObjectInspector);
        fieldOIs.add(PrimitiveObjectInspectorFactory.writableStringObjectInspector);

        // 将返回值的列描述信息进行打包并创建StructObjectInspector返回
        return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames, fieldOIs);
    }

    /**
     * 逻辑处理：
     *      args[0]: [{"movie_name":"肖申克的救赎","MovieType":"犯罪"},{"movie_name":"肖申克的救赎","MovieType":"剧情"}]
     * @param args
     * @throws HiveException
     */
    @Override
    public void process(Object[] args) throws HiveException {

        try {
            // 由于传入的字符串为JSON数组，那么需要将其包装成JSONArray
            JSONArray jsonArray = new JSONArray(args[0].toString());
            // 拿到数组对象之后，需要循环取其中的JSON数据
            for (int i = 0; i < jsonArray.length(); i++) {
                // 取出JSON数据后，对应用JSONObject对象存储具体的数据 {"movie_name":"肖申克的救赎","MovieType":"犯罪"}
                JSONObject jsonObject = jsonArray.getJSONObject(i);
                String movie_name = jsonObject.getString("movie_name");
                String movie_type = jsonObject.getString("MovieType");

                // 取出数据之后，可以按行写出
                Text[] oneLine = new Text[2];
                oneLine[0] = new Text(movie_name);
                oneLine[1] = new Text(movie_type);
                forward(oneLine);

            }

        } catch (JSONException e) {
            e.printStackTrace();
        }

    }

    @Override
    public void close() throws HiveException {

    }

    /**
     * 作业：
     *  对于 {"movie": [{"movie_name": "肖申克的救赎", "MovieType": "犯罪" }, {"movie_name": "肖申克的救赎", "MovieType": "剧情" }]}
     *  数据，将数据直接传入explodeMovie函数中，可以得到 和上述代码一致的结果
     *  SELECT explodeMovie(movie)  FROM moives
     *
     *  肖申克的救赎	犯罪
     *  肖申克的救赎	剧情
     */
}
